// rabbitmq package supports automaticaly reconncet rabbit and recreate publishor
package rabbitmq

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/rs/zerolog/log"
	"github.com/streadway/amqp"
)

// Publisher can publish msg to rabbitmq
type Publisher struct {
	ExchangeName string
	BindKey      string
	Channel      *amqp.Channel
	bindQueue    BindQueueFunc
}

// newPublisher returns a new Publishor instance.
func newPublisher(ch *amqp.Channel, exchangeName, bindKey string, bindQueue BindQueueFunc) *Publisher {
	return &Publisher{
		ExchangeName: exchangeName,
		BindKey:      bindKey,
		Channel:      ch,
		bindQueue:    bindQueue,
	}
}

// Publish publishs msg to declared queue.
func (p *Publisher) Publish(msg []byte) error {
	if len(msg) == 0 {
		return nil
	}
	return p.Channel.Publish(p.ExchangeName, p.BindKey, false, false,
		amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			MessageId:    uuid.NewString(),
			Timestamp:    time.Now(),
			Body:         msg,
		},
	)
}

// BindQueueFunc is a func that handle bind queue and exchange on channel
type BindQueueFunc func(p *Publisher) error

// DefaultBindQueueFunc 获取指定队列名称的默认bindQueueFunc
var DefaultBindQueueFunc = func(queueName string) BindQueueFunc {
	return func(p *Publisher) error {
		err := p.Channel.ExchangeDeclare(p.ExchangeName, "direct",
			true, false, false, false, nil)
		if err != nil {
			return err
		}

		queue, err := p.Channel.QueueDeclare(queueName, true, false, false, false, nil)
		if err != nil {
			return err
		}

		err = p.Channel.QueueBind(queue.Name, p.BindKey, p.ExchangeName, false, nil)
		if err != nil {
			return err
		}

		return nil
	}
}

// pc will be assigned by Start.
var pc *publisherContainer

// publisherContainer is a container that cache these created Publishers.
// When the amqp.Connection which created with connection string endpoint was exception,
// the container will automate refresh(recreate) the Publishers in it.
type publisherContainer struct {
	endpoint   string
	mux        sync.RWMutex
	amqpConn   *amqp.Connection
	publishers map[string]*Publisher
}

// NewPublisherContainer will create a new PublisherContainer instance,and will directly start reconnection automaticly.
// Usage:
// 		go NewPublisherContainer(ctx,endpoint,errChan)
//
//
//	In other code snip where require use the Publisher by uniquePublisherName:
// 		Container.DeclarePublisher(uniquePublisherName) can declare and store the Publisher
//		p,ok:=Container.Get(uniquePublisherName)
//		if !ok{
//		}
//		err:=p.Pulish(msg)
func NewPublisherContainer(ctx context.Context, endpoint string, errChan chan<- error) {
	amqpConn, err := amqp.Dial(endpoint)
	if err != nil {
		errChan <- err
		return
	}
	pc = &publisherContainer{
		endpoint:   endpoint,
		amqpConn:   amqpConn,
		publishers: make(map[string]*Publisher),
	}

	go start(ctx, errChan)
}

// start starts the automaticaly refresh logic in blocking way.
func start(ctx context.Context, errChan chan<- error) {
	subCtx, canceled := context.WithCancel(ctx)
	rabbitErrChan := pc.amqpConn.NotifyClose(make(chan *amqp.Error))
	for {
		select {
		case <-ctx.Done():
			pc.amqpConn.Close()
			canceled()
			log.Error().Msg("daemon server closed: normal")
			return
		case e, ok := <-rabbitErrChan:
			log.Error().Err(e).Bool("ok", ok).Msg("rabbitmq connection close")
			if !ok { //正常退出
				canceled()
				errChan <- e
				return
			}
			amqpConn, err2 := amqp.Dial(pc.endpoint)
			if err2 != nil {
				log.Error().Err(err2).Msg("try reconnection rabbit err")
				time.Sleep(10 * time.Second)
				continue
			}
			pc.amqpConn = amqpConn
			canceled()
			subCtx, canceled = context.WithCancel(ctx)
			log.Error().Msg("rabbitmq try reconnection OK")
			err2 = pc.refresh(subCtx)
			if err2 != nil {
				log.Error().Err(err2).Msg("rabbitmq refresh err")
				time.Sleep(10 * time.Second)
				continue
			}
		}
	}
}

// DeclarePublisher declares and stores the publisher.
// If the Publisher has existed,will return directly.
func DeclarePublisher(uniquePublisherName, exchangeName, bindKey string, bindQueue BindQueueFunc) error {
	if pc == nil {
		return errors.New("rabbitmq does not be initialized")
	}
	pc.mux.Lock()
	defer pc.mux.Unlock()
	_, ok := pc.publishers[uniquePublisherName]
	if ok {
		return nil
	}
	p, err := pc.createPublisher(exchangeName, bindKey, bindQueue)
	if err != nil {
		return err
	}
	pc.publishers[uniquePublisherName] = p
	return nil
}

// Get gets the uniquePublisherName's Publishor
func Get(uniquePublisherName string) (p *Publisher, ok bool) {
	pc.mux.RLock()
	defer pc.mux.RUnlock()
	p, ok = pc.publishers[uniquePublisherName]
	return
}

func (pc *publisherContainer) createPublisher(exchangeName, bindKey string, bindQueue BindQueueFunc) (*Publisher, error) {
	ch, err := pc.amqpConn.Channel()
	if err != nil {
		log.Error().Err(err).Msg("rabbitmq create channel err")
		return nil, err
	}
	p := newPublisher(ch, exchangeName, bindKey, bindQueue)
	err = bindQueue(p)
	if err != nil {
		return nil, err
	}
	return p, nil
}

func (pc *publisherContainer) refresh(ctx context.Context) error {
	pc.mux.Lock()
	defer pc.mux.Unlock()
	for k, v := range pc.publishers {
		p, err := pc.createPublisher(v.ExchangeName, v.BindKey, v.bindQueue)
		if err != nil {
			log.Error().Err(err).Msg("rabbitmq createPublisher err")
			return err
		}
		v.Channel.Close()
		pc.publishers[k] = p
	}
	return nil
}
